package com.hwn.datax.service.impl;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.lang.UUID;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.hwn.datax.domain.DataxJob;
import com.hwn.datax.domain.DataxJobRunLog;
import com.hwn.datax.domain.ServerInfo;
import com.hwn.datax.domain.datax.RangeOption;
import com.hwn.datax.service.ConfigService;
import com.hwn.datax.service.DataxJobService;
import com.hwn.datax.service.ServerInfoService;
import com.hwn.datax.utils.JschUtil;
import com.hwn.datax.utils.RegexUtil;
import com.hwn.datax.utils.Result;
import com.hwn.datax.utils.SplitUtil;
import com.hwn.datax.utils.enums.RESULT_CODE;
import javafx.application.Platform;
import javafx.collections.FXCollections;
import javafx.collections.ObservableList;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import java.time.LocalDate;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author Administrator
 * @description 针对表【datax_job(datax任务)】的数据库操作Service实现
 * @createDate 2023-06-04 17:00:35
 */
@Service
public class DataxJobServiceImpl implements DataxJobService {


    ServerInfoService serverInfoService;

    ConfigService configService;

    private static final String CONFIG_KEY = "DataxJobTable";

    private final ObservableList<DataxJobRunLog> dataxJobRunLogList = FXCollections.observableArrayList();


    @Override
    public ObservableList<DataxJobRunLog> getDataxJobRunLogList() {
        return dataxJobRunLogList;
    }

    @Autowired
    public DataxJobServiceImpl(ServerInfoService serverInfoService, ConfigService configService) {
        this.serverInfoService = serverInfoService;
        this.configService = configService;
        JSONObject config = configService.getConfig(CONFIG_KEY);
        if (config != null) {
            config.forEach((k, v) -> {
                dataxJobMap.put(Convert.toLong(k), JSONUtil.toBean(v.toString(), DataxJob.class));
            });
        }
//        initDataxJobRunLogList();
    }

    private final Map<Long, DataxJob> dataxJobMap = new HashMap<>();

    /**
     * 上传job到datax服务器
     *
     * @param id dataxJobId
     * @return 上传后的路径
     */
    @Override
    public DataxJob uploadJson(Long id) {
        DataxJob dataxJob = dataxJobMap.get(id);
        Assert.notNull(dataxJob, "根据传入的jobId{" + id + "}，未获取到对应的job，请检查！");
        return dataxJob;
    }

    /**
     * 上传job到datax服务器
     *
     * @param dataxJob dataxJob对象
     * @return
     */
    @Override
    public DataxJob uploadDataxJob(DataxJob dataxJob) {
        Integer serverId = dataxJob.getServerId();
        ServerInfo serverInfo = serverInfoService.getById(serverId);
        String dataxJsonPath = serverInfo.getDataxJsonPath();
        dataxJsonPath = dataxJsonPath.endsWith("/") ? dataxJsonPath : dataxJsonPath + "/";
        String path = dataxJsonPath + dataxJob.getUuid() + ".json";
        dataxJob.setJsonPath(path);
        JschUtil.uploadFile(serverInfo, path, dataxJob.getJobParams());
        return dataxJob;
    }


    /**
     * 获取分组下的所有任务
     *
     * @param groupId 分组ID
     * @return 任务列表
     */
    @Override
    public List<DataxJob> listByGroupId(Long groupId) {
        return dataxJobMap.values().stream().filter(t -> groupId.equals(t.getGroupId())).collect(Collectors.toList());
    }

    /**
     * 后台执行job任务
     *
     * @param id job主键
     * @return 返回任务执行标识uuid
     */
    @Override
    public Result<String> executeDataxJob(Long id) {
        DataxJob dataxJob = uploadJson(id);
        ArrayList<DataxJob> list = new ArrayList<>();
        list.add(dataxJob);
        String uuid = executeDataxJobList(list, 0);
        return new Result<String>(RESULT_CODE.SUCCESS.getCode(), uuid, uuid);
    }

    /**
     * 切分执行任务
     *
     * @param id 任务id
     */
    @Override
    public void executeDataxJobByRange(Long id) {
        DataxJob dataxJob = dataxJobMap.get(id);
        Assert.notNull(dataxJob, "根据传入的jobId{" + id + "}，未获取到对应的job，请检查！");
        List<DataxJob> dataxJobs = setSplit(dataxJob);
        executeDataxJobList(dataxJobs, 0);
    }

    /**
     * 后台执行job任务
     *
     * @param list  job列表
     * @param index 任务序号
     * @return 返回任务执行标识uuid
     */
    private String executeDataxJobList(List<DataxJob> list, Integer index) {
        Assert.notEmpty(list, "传入的job列表不能为空！");
        String uuid = UUID.randomUUID().toString();
        list.get(index).setUuid(uuid);
        DataxJob dataxJob = uploadDataxJob(list.get(index));
        Integer serverId = dataxJob.getServerId();
        ServerInfo serverInfo = serverInfoService.getById(serverId);
        String shellCommand = buildShellCommand(serverInfo, dataxJob);
        DataxJobRunLog jobRunLog = new DataxJobRunLog();

        Platform.runLater(() -> {
            jobRunLog.setUuid(uuid);
            jobRunLog.setJobId(dataxJob.getId());
            String jobName = dataxJob.getJobName();
            if (!list.isEmpty() && list.size() > 1) {
                jobName += "(" + (index + 1) + "/" + list.size() + ")";
            }
            jobRunLog.setJobName(jobName);
            jobRunLog.setJobPath(dataxJob.getJsonPath());
            jobRunLog.setShellCommand(shellCommand);
            jobRunLog.setServerName(serverInfo.getServerName());
            jobRunLog.setServerIp(serverInfo.getServerIp());
            jobRunLog.setLogPath(dataxJob.getLogPath());
            jobRunLog.setServerId(dataxJob.getServerId());
            jobRunLog.setStartTime(new Date());
            jobRunLog.setStatus(0);
            jobRunLog.setTotal(list.size());
            jobRunLog.setCurrent(index + 1);
            dataxJobRunLogList.add(jobRunLog);
        });

        StringBuilder logContent = new StringBuilder();
        JschUtil.asyncExecCommand(serverInfo, shellCommand, (msg) -> {
            logContent.append(msg).append("\r\n");

            //匹配进度百分比
            String percentageString = RegexUtil.match("Percentage (\\d+\\.\\d+)%", msg);

            //匹配记录数
            String totalRecordsString = RegexUtil.match("Total (\\d+) records", msg);

            Platform.runLater(() -> {
                jobRunLog.setLogContent(logContent.toString());
                if (null != percentageString) {
                    Double percentage = Convert.toDouble(percentageString);
                    jobRunLog.setPercentage(percentage);
                }
                if (null != totalRecordsString) {
                    Long totalRecords = Convert.toLong(totalRecordsString);
                    jobRunLog.setTotalRecords(totalRecords);
                }
            });
        }, (msg) -> {
            Platform.runLater(() -> {
                //mgs 0表示成功，其它都是失败
                jobRunLog.setStatus(msg.equals(0) ? 1 : 2);
                jobRunLog.setResult(msg);
                jobRunLog.setEndTime(new Date());
            });
            //如果执行成功并且还有剩余任务，则继续执行
            if (msg.equals(0)) {
                if (index < list.size() - 1) {
                    jobRunLog.setHasNext(true);
                    jobRunLog.setNextUuid(executeDataxJobList(list, index + 1));
                } else {
                    jobRunLog.setHasNext(false);
                }
            }
        });
        return uuid;
    }


    /**
     * 设置分段执行
     *
     * @param dataxJob 任务对象
     */
    private List<DataxJob> setSplit(DataxJob dataxJob) {
        if (dataxJob.getRangeOption() == null || dataxJob.getRangeOption().getType() == null) {
            throw new RuntimeException("分段执行选项格式错误");
        }
        List<DataxJob> result = new ArrayList<>();
        RangeOption rangeOption = dataxJob.getRangeOption();


        String type = rangeOption.getType();
        String replaceStr = rangeOption.getReplaceStr();

        List<List<String>> lists = null;
        if (type.equals("range")) {
            Long start = rangeOption.getStartValue();
            Long end = rangeOption.getEndValue();
            Long step = rangeOption.getStep();
            lists = SplitUtil.splitRange(start, end, step);
        }
        if (type.equals("date")) {
            LocalDate startDate = rangeOption.getStartDate();
            LocalDate endDate = rangeOption.getEndDate();
            Long step = rangeOption.getStep();
            lists = SplitUtil.splitDate(startDate, endDate, Convert.toInt(step));
        }
        if (lists != null) {
            for (int i = 0; i < lists.size(); i++) {
                List<String> list = lists.get(i);
                result.add(copyAndSetJobReaderWhere(dataxJob, replaceStr, list, i > 0));
            }
        }
        return result;
    }


    /**
     * 复制并设置where条件
     *
     * @param dataxJob   原始job
     * @param replaceStr 替换的字符串
     * @param params     替换的参数
     * @param isNext     是否是下一个任务,用来去掉 truncate table 的内容
     * @return 返回一个新的job任务对象
     */
    private DataxJob copyAndSetJobReaderWhere(DataxJob dataxJob, String replaceStr, List<String> params, Boolean isNext) {
        DataxJob curr = new DataxJob();
        BeanUtils.copyProperties(dataxJob, curr);
        JSONObject obj = JSONUtil.parseObj(curr.getJobParams());
        JSONObject parameter = obj.getJSONObject("job").
                getJSONArray("content").
                getJSONObject(0).
                getJSONObject("reader").
                getJSONObject("parameter");
        String where;
        if (parameter.containsKey("where") && null != parameter.getStr("where")) {
            where = parameter.getStr("where");
        } else {
            where = "";
        }
        if (StrUtil.isNotBlank(where)) {
            where += " and ";
        }
        replaceStr = replaceStr.replace("#{0}", params.get(0));
        replaceStr = replaceStr.replace("#{1}", params.get(1));
        where = where + replaceStr;
        parameter.set("where", where);

        if (isNext) {
            JSONObject writerParameter = obj.getJSONObject("job").
                    getJSONArray("content").
                    getJSONObject(0).
                    getJSONObject("writer").
                    getJSONObject("parameter");
            JSONArray preSql = writerParameter.getJSONArray("preSql");
            if (writerParameter.containsKey("preSql") && preSql != null && !preSql.isEmpty()) {
                preSql.removeIf(t -> {
                    String sql = (String) t;
                    return sql.toUpperCase().contains("TRUNCATE")
                            && sql.toUpperCase().contains(curr.getWriter().getTableName().toUpperCase())
                            && sql.toUpperCase().contains("TABLE")
                            && curr.getWriter().getTruncateTable();

                });
            }
        }


        curr.setJobParams(JSONUtil.toJsonStr(obj));
        return curr;
    }

    /**
     * 结束任务
     *
     * @param uuid 任务标识
     * @return
     */
    @Override
    public Boolean killDataxJob(String uuid) {
        DataxJobRunLog execLog = getExecLogByKey(uuid);
        String jobPath = execLog.getJobPath();
        Integer serverId = execLog.getServerId();
        ServerInfo serverInfo = serverInfoService.getById(serverId);
        String command = "pkill -f -9  \"" + jobPath + "\"";
        JschUtil.execCommand(serverInfo, command);
        return true;
    }

    /**
     * 获取任务执行情况
     *
     * @param uuid 任务标识
     * @return
     */
    @Override
    public DataxJobRunLog getExecLogByKey(String uuid) {
        return dataxJobRunLogList.stream().filter(t -> t.getUuid().equals(uuid)).findFirst().get();
    }

    @Override
    public Long removeExecKeys(List<String> keys) {
        Platform.runLater(() -> {
            keys.forEach(uuid -> {
                DataxJobRunLog execLog = getExecLogByKey(uuid);
                execLog.statusProperty().removeAllMyListeners();
                dataxJobRunLogList.remove(execLog);
            });
        });
        return (long) keys.size();
    }

    /**
     * 构建datax执行命令
     *
     * @param serverInfo datax服务器
     * @param dataxJob   dataxJob对象
     * @return 返回命令语句
     */
    @Override
    public String buildShellCommand(ServerInfo serverInfo, DataxJob dataxJob) {
        StringBuilder sb = new StringBuilder();
        sb.append("python ");
        String dataxPath = serverInfo.getDataxPath();
        dataxPath = dataxPath.endsWith("/") ? dataxPath : dataxPath + "/";
        sb.append(dataxPath);
        sb.append("bin/datax-my.py");
        String uuid = dataxJob.getUuid();
        //指定日志文件名称
        sb.append(" --log ").append(uuid);
        dataxJob.setLogPath(dataxPath + "log/" + DateUtil.format(new Date(), DatePattern.NORM_DATE_PATTERN) + "/" + uuid + ".log");
        //设置jvm参数
        if (StrUtil.isNotBlank(dataxJob.getJvmParams())) {
            sb.append(" --jvm=\"");
            sb.append(dataxJob.getJvmParams());
            sb.append("\"");
        }
        sb.append(" ").append(dataxJob.getJsonPath());
        return sb.toString();
    }

    /**
     * 保存任务
     *
     * @param dataxJob
     * @return
     */
    @Override
    public DataxJob save(DataxJob dataxJob) {
        if (dataxJob.getId() == null) {
            dataxJob.setId(IdUtil.getSnowflakeNextId());
        }
        dataxJobMap.put(dataxJob.getId(), dataxJob);
        configService.saveConfig(CONFIG_KEY, JSONUtil.toJsonStr(dataxJobMap));
        return dataxJob;
    }

    @Override
    public void remove(Long jobId) {
        dataxJobMap.remove(jobId);
        configService.saveConfig(CONFIG_KEY, JSONUtil.toJsonStr(dataxJobMap));
    }
}




